package com.web.webtitle.dao;

import com.web.webtitle.entity.TitleHot;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.stereotype.Repository;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

@Repository
public class HBaseDao {
    //获取连接
    public Connection getConnection() throws IOException {
        //设置本机客户端访问路径
        System.setProperty("hadoop.home.dir","E:\\hadoop");
        //设置访问身份
        System.setProperty("HADOOP_USER_NAME","root");
        //获取配置对象
        Configuration config = HBaseConfiguration.create();
        //hbase客户端配置zookeeper的集群地址
        config.set("hbase.zookeeper.quorum", "master,slave1,slave2");
        //设置端口号
        config.set("hbase.zookeeper.property.clientPort", "2181");
       return ConnectionFactory.createConnection(config);
    }

    //时间段查询
    public List<TitleHot> query(long start,long end) throws IOException {
        Connection connection = this.getConnection();
        //获取table对象 操作table
        Table table = connection.getTable(TableName.valueOf("title_hot"));
        //获取扫描对象
        Scan scan = new Scan();
        //得到某一个时间段的数据
        scan.setTimeRange(start,end);
        //获取迭代器
        Iterator<Result> iterator = table.getScanner(scan).iterator();
        List<TitleHot> titleHots = new ArrayList<>();
        //遍历
        while (iterator.hasNext()){
            //获取下一行的数据
            Result result = iterator.next();//行数据
            //单元格
            List<Cell> cells = result.listCells();
            TitleHot hot = new TitleHot();
            for (Cell cell : cells) {
                hot.setId(Bytes.toString(CellUtil.cloneRow(cell)));
                String col = Bytes.toString(CellUtil.cloneQualifier(cell));
                String value = Bytes.toString(CellUtil.cloneValue(cell));
                if ("userName".equals(col)){
                    hot.setUserName(value);
                }else if ("agentType".equals(col)){
                    hot.setAgentType(value);
                }else if ("time".equals(col)){
                    hot.setTime(value);
                }else if ("title".equals(col)){
                    hot.setTitle(value);
                }

            }
            //装载行信息
            titleHots.add(hot);
        }
        table.close();
        connection.close();
        return titleHots;
    }
}
